package com.batchengine.worker;

import static com.batchengine.Constants.MILLIS_IN_A_SECOND;
import com.batchengine.task.TaskContext;
import static com.batchengine.task.TaskUtil.getTaskContext;
import static com.batchengine.util.LoggingFactory.createLogger;
import static com.batchengine.worker.Status.*;
import org.apache.log4j.Logger;

import static java.lang.Thread.*;
import static java.text.MessageFormat.format;

/**
 * Created by sunil.incubator@yahoo.com.
 * User: sunil.kalva
 * Date: Jun 22, 2009
 */
public abstract class Worker implements Runnable {

    private static final Logger logger = createLogger();

    private static Long counter = 1l;

    private String name;
    private Long jobId;
    private Long workerId;

    private Status status = READY;

    /**
     * The thread priority for this worker.
     */
    protected int workerPriority = NORM_PRIORITY;

    /**
     * The thread group that this worker belongs to.
     */
    protected ThreadGroup workGroup;

    /**
     * A reference to the workers executing Thread.
     */
    protected Thread workerThread;

    /**
     * An object used for internal synchronization.
     */
    protected Object[] internalLock = new Object[0];

    private long delayInSeconds = 1;

    public Worker(String name, long jobId) {
        this.name = name;
        this.jobId = jobId;
        this.workerId = counter++;
        setStatus(READY);
    }

    protected Worker(String name, long jobId, ThreadGroup workGroup) {
        this.name = name;
        this.workGroup = workGroup;
        this.jobId = jobId;
        this.workerId = counter++;
    }

    protected Worker(String name, long jobId, ThreadGroup workGroup, int workerPriority) {
        this.name = name;
        this.workerPriority = workerPriority;
        this.workGroup = workGroup;
        this.jobId = jobId;
        this.workerId = counter++;
    }

    public Long getWorkerId() {
        return workerId;
    }

    public Long getJobId() {
        return jobId;
    }

    public void setJobId(Long jobId) {
        this.jobId = jobId;
    }

    protected long getDelayInSeconds() {
        return delayInSeconds;
    }

    public void setDelayInSeconds(long delayInSeconds) {
        this.delayInSeconds = delayInSeconds;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    protected void setStatus(Status status) {
        this.status = status;
    }

    public Status getStatus() {
        return status;
    }

    /**
     * This is the method used to start the worker thread.
     */
    public synchronized void start() {

        if (this.workGroup != null && this.name != null) {
            workerThread = new Thread(this.workGroup, this, this.name);
        } else if (this.workGroup != null) {
            workerThread = new Thread(this.workGroup, this);
        } else if (this.name != null) {
            workerThread = new Thread(this, this.name);
        } else {
            workerThread = new Thread(this);
        }

        this.workerPriority = Math.max(MIN_PRIORITY, Math.min(MAX_PRIORITY, this.workerPriority));
        this.name = workerThread.getName();
        workerThread.setPriority(this.workerPriority);
        workerThread.start();
    }

    public synchronized void stop() {

        try {

            this.status = STOPPED;
            while (RUNNING.equals(getStatus())) {
                internalLock.wait(3 * MILLIS_IN_A_SECOND);
            }
        } catch (InterruptedException ignored) {
            logger.error(format("error stopping the worker {0}", getName()));
            //do nothing
        }
    }

    public synchronized void pause() {

//        if (Status.RUNNING.equals(getStatus())) {
//            return;
//        }

        try {

            this.status = PAUSED;
            while (RUNNING.equals(getStatus())) {
                internalLock.wait(3 * MILLIS_IN_A_SECOND);
            }
        } catch (InterruptedException ignored) {
            logger.error(format("error stopping the worker {0}", getName()));
            //doin nothing
        }
    }

    public void resume() {

        if (PAUSED.equals(getStatus())) {
            status = RUNNING;
        }
    }

    /**
     * This method is called before the thread stops execution.
     * The default implementation does nothing. It can be overridden
     * by the deriving class if desired.
     */
    protected void cleanUp() {
        //nothing
    }

    public void run() {

        try {

            this.status = RUNNING;
            workerThread = currentThread();
            TaskContext taskContext = getTaskContext(getJobId());
            while (!STOPPED.equals(status)) {

                sleep(1000 * delayInSeconds);
                while (!PAUSED.equals(status) && !STOPPED.equals(status)) {

                    if (!(this instanceof DataGatherer)) {
                        sleep(1000 * delayInSeconds);
                    } else {
                        sleep(500);
                    }
                    doWork(taskContext);
                }
                if (PAUSED.equals(status)) {
                    logger.debug(String.format("=====> %s : PAUSED <=====", this.getName()));
                }
            }
        } catch (InterruptedException ignored) {
            logger.error(format("error running the worker {0}", workerThread.getName()), ignored);
            //do nothing - if stop is true then we will exit.
        } finally {
            cleanUp(); // unnecessary but can help GC
            this.status = FINISHED;
        }
    }

    public abstract WorkerType getType();

    /**
     * The method that does all the work. This is implemented by the
     * deriving class.
     *
     * @throws InterruptedException throws InterruptedException
     */
    protected abstract void doWork(TaskContext context) throws InterruptedException;

    @Override
    public String toString() {
        return "Worker{" +
                "name='" + name + '\'' +
                ", jobId=" + jobId +
                ", workerId=" + workerId +
                '}';
    }
}
